package com.mtiiot.fsmanage.server;

import com.mtiiot.service.BaseService;
import com.mtiiot.fsmanage.bean.common.SysTreeNode;
import com.mtiiot.fsmanage.bean.common.TableDataInfo;
import com.mtiiot.fsmanage.core.dao.ElasticsearchDataSource;
import com.mtiiot.fsmanage.utils.Contants;
import com.mtiiot.fsmanage.utils.FsViewConfig;
import com.mtiiot.fsmanage.utils.TimeUtil;
import com.mtiiot.fscore.bean.DfsFileInfo;
import com.mtiiot.fscore.bean.MetadataInfo;
import com.mtiiot.fscore.inter1.FastdfsService1;
import com.mtiiot.fscore.utils.StringUtil;
import com.mtiiot.fsmanage.core.dao.BaseElasticsearch;
import net.sf.json.JSONArray;
import net.sf.json.JSONObject;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchType;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.UnsupportedEncodingException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * Created by xyr on 2016/8/24.
 * <p/>
 * netty 接口查询服务类
 */

@Service
@Transactional
public class NettyServer {

    @Autowired
    private BaseElasticsearch baseElasticsearch;
    private static String INDEX = Contants.INDEX_METADATA;
    private static String TYPE = Contants.TYPE_METADATA_FASTDFS;
    private static String SYSNAME_NETTY = Contants.FASTDFS_FILE_TYPE_NETTY;

    @Autowired
    private FastdfsService1 fastdfsService1;
    @Autowired
    private FsViewConfig fsViewConfig;

    @Autowired
    private ElasticsearchDataSource esDataSource;
    public ElasticsearchDataSource getEsDataSource() {
        return esDataSource;
    }
    public void setEsDataSource(ElasticsearchDataSource esDataSource) {
        this.esDataSource = esDataSource;
    }



    /**
     * 文件查询
     */
    public Object nettyQuery(String sysName, String deviceId, String code, String times, String queryKeyWord, int count) throws Exception {
        Client client = null;
        try {
            client = baseElasticsearch.getEsDataSource().getClient();
            BoolQueryBuilder bq = QueryBuilders.boolQuery();
            // 默认查询netty系统的数据
            bq.must(QueryBuilders.termQuery("sysName", SYSNAME_NETTY));
            if (!StringUtil.isEmpty(code)) {
                bq.must(QueryBuilders.termQuery("deviceId", code));                 // Query
            }
            if (!StringUtil.isEmpty(deviceId)) {
                bq.must(QueryBuilders.termQuery("code", deviceId.toLowerCase()));                 // Query
            }

            if (!StringUtil.isEmpty(queryKeyWord)) {
                bq.must(QueryBuilders.queryStringQuery(queryKeyWord));
            }
            long[] l = TimeUtil.handleTimes(times);
            if (l != null && l.length > 0) {
                bq.must(QueryBuilders.rangeQuery("time").from(l[0]).to(l[1]));
            }
            QueryBuilder queryBuilder = bq;
            SearchRequestBuilder srb = client.prepareSearch(INDEX)
                    .setTypes(TYPE)
                            //按时间降序号
                    .addSort("time", SortOrder.DESC)
                    .setSearchType(SearchType.DFS_QUERY_THEN_FETCH)
                            // 设置查询数据的位置,分页用
                    .setFrom(0)
                            // 设置查询结果集的最大条数
                    .setSize(count)
                            // 设置是否按查询匹配度排序
                    .setExplain(true);
            // 最后就是返回搜索响应信息
            srb.setQuery(bq);

            TableDataInfo tdif=new TableDataInfo();
            SearchResponse response =srb.execute().actionGet();
            SearchHits searchHits = response.getHits();
            tdif.setCount(searchHits.totalHits());
            SearchHit[] hits = searchHits.getHits();

            List<Map<String, Object>> sms=new ArrayList<Map<String, Object>>();
            for (SearchHit searchHit : hits) {
                Map<String, Object> sm = searchHit.sourceAsMap();
                sms.add(sm);
            }
            tdif.setObject(sms);
            return tdif;
        } finally {
            if (client != null) {
                baseElasticsearch.getEsDataSource().releaseClient(client);
            }
        }
    }

    /**
     * 获得fastdfs 文件系统的树
     */
    public Object getNettyChildrenNode(String id, String deviceId) throws Exception {
        Client client = null;
        try {
            client = baseElasticsearch.getEsDataSource().getClient();
            SearchResponse sr = null;
            List<Terms.Bucket> list = null;
            List<SysTreeNode> sts = new ArrayList<SysTreeNode>();
            boolean hasChildren = true;
            if (id.equals("#")) {
                //////////////////////查询系统////////////////
                sr = client.prepareSearch(INDEX)
                        .setTypes(TYPE)
                        .setSize(0)
                        .setQuery(QueryBuilders.matchAllQuery())
                        .setQuery(QueryBuilders.termQuery("sysName", SYSNAME_NETTY))                 // Query
                        .addAggregation(
                                AggregationBuilders.terms("agg1").field("code").size(500)
                        ).execute().actionGet();

            } else if (id.split("-").length == 2) {
                //////////////////////查询L1////////////////
                hasChildren=false;
                sr = client.prepareSearch(INDEX)
                        .setTypes(TYPE)
                        .setSize(0)
                        .setQuery(QueryBuilders.matchAllQuery())
                        .setQuery(QueryBuilders.termQuery("sysName", SYSNAME_NETTY))                 // Query
                        .setQuery(QueryBuilders.termQuery("code", deviceId.toLowerCase()))                 // Query
                        .addAggregation(
                                AggregationBuilders.terms("agg1").field("deviceId").size(500)
                        ).execute().actionGet();
            }
            Terms agg1 = sr.getAggregations().get("agg1");
            list = agg1.getBuckets();
            int i = 1;
            Object key = null;
            String text = null;
            for (Terms.Bucket b : list) {
                SysTreeNode st = new SysTreeNode();
                st.setId(id + "-" + i);
                key = b.getKey();
                text = key.toString();
                st.setText(text);
                st.setChildren(hasChildren);
                sts.add(st);
                i++;
            }
            return sts;
        } finally {
            if (client != null) {
                baseElasticsearch.getEsDataSource().releaseClient(client);
            }
        }
    }

    /**
     * 根据文件ID获得分布式文件的内容
     */
    public Object getFileContentById(String dfsId, String code) throws UnsupportedEncodingException {
        Map<String, String> map = new HashMap<String, String>();
        //String id = "group1/M00/1D/41/CnXXwVe7uHSAIiWhAABELa0_rJs97.docx";
        DfsFileInfo fileInfo = fastdfsService1.getFile(new MetadataInfo(fsViewConfig.getFsviewSysId()), dfsId);
        if (fileInfo == null) return map;
        String msg = new String(fileInfo.getFile(), "utf-8");
        map.put("flag", "true");
        map.put("msg", msg);
        return map;
    }

    /**
     * 将选中的记录重新入库解析
     */
    public Object uploadNettyFiles(String nettyFilesJson) throws Exception {
        JSONArray jsonArray = JSONArray.fromObject(nettyFilesJson);
        int len = jsonArray.size();
        // 返回页面用到
        Map<String, String> map = new HashMap<String, String>();
        // 修改重传时间用到
        // 批量更新map map结构中 3个string分别对应(id,field,value)
        Map<String,Map<String,String>> editMap = new HashMap<String, Map<String, String>>();

        // 正常的重传方式，在分布式问加你系统取出数据
        for (int i = 0; i < len; i++) {
            String json = jsonArray.getString(i);
            JSONObject jsonObject = JSONObject.fromObject(json);
            String dfsId = jsonObject.getString("dfsId");
            String code = jsonObject.getString("code").toUpperCase().replaceAll("-","_");
            String seqNo = jsonObject.getString("seqNo");
            if (StringUtil.isEmpty(code) || StringUtil.isEmpty(seqNo) || StringUtil.isEmpty(dfsId)) continue;
            // 根据code判断数据是否可以重传
            //if (!Boolean.valueOf(codeMap.get(code.toUpperCase()))) continue;
            if (code.toUpperCase().equals("N_BX_STA_HSZZ")) continue;
            // 重传,调用入库的接口
            BaseService.sendData(code, seqNo, dfsId);
            // 重传成功，修改索引的
            // 获得Id,修改重传次数和时间
            String id = jsonObject.getString("id");
            // 获得重传的次数
            int retransmissionCount_int = jsonObject.getInt("retransmissionCount");
            int retransmissionCount = retransmissionCount_int + 1;
            // 修改重传的时间
            String retransmissionTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
            Map<String,String> valueMap = new HashMap<String, String>();
            valueMap.put("retransmissionCount",String.valueOf(retransmissionCount));
            valueMap.put("retransmissionTime",retransmissionTime);
            editMap.put(id,valueMap);
        }

        // 批量更新
        if (editMap.size() > 0){
            batchUpdateIndex(INDEX,TYPE,editMap);
        }
        map.put("total", String.valueOf(len));
        map.put("msg", "success");
        return map;
    }

    /**
     * 批量修改文档
     * @param index 索引名称
     * @param type  索引type
     * @param map  批量更新map map结构中 3个string分别对应(id,field,value)
     */
    public boolean batchUpdateIndex(String index,String type,Map<String,Map<String,String>> map) throws Exception {
        Client client=null;
        try{
            client=esDataSource.getClient();
            BulkRequestBuilder bulkRequest=client.prepareBulk();
            Set<String> keys=map.keySet();
            for(String key:keys){
                bulkRequest.add(client.prepareUpdate(index, type,key).setDoc(map.get(key)));
            }
            BulkResponse bulkResponse=bulkRequest.execute().actionGet();
            boolean result=bulkResponse.hasFailures();
            if(result){
                throw new Exception(bulkResponse.buildFailureMessage());
            }
            return !result;
        }finally {
            if(client!=null){
                esDataSource.releaseClient(client);
            }
        }
    }

}
